package com.sduept.nwld.wave.traveling;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.annotation.PostConstruct;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.monitor.FileAlterationListener;
import org.apache.commons.io.monitor.FileAlterationListenerAdaptor;
import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.sduept.bigdata.fault.entity.TravelingWaveFaultRecord;
import com.sduept.bigdata.fault.system.properties.FaultSystemProperties;
import com.sduept.bigdata.weather.map.MapDataCache;
import com.sduept.cim.model.core.Substation;
import com.sduept.cim.model.extend.LineSegment;
import com.sduept.cim.model.protection.TravelingWaveEquipment;
import com.sduept.cim.model.wires.ACLineSegment;
import com.sduept.cim.model.wires.DcLinesegment;
import com.sduept.cim.query.service.ACLineSegmentQueryService;
import com.sduept.cim.query.service.SubstationQueryService;
import com.sduept.core.dao.SuperDAO;
import com.sduept.core.entity.IdentifiedObject;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author 行波监控服务，该功能会遇到以下几个问题 1、服务器停止，但是监控目录在这期间推送了部分数据，这些数据是未解析同步到库中的数据
 *         2、服务正常运行，推送过来的数据，需要及时解析处理，防止拥塞 3、数据可能存在多次处理，需要做好幂等处理
 *         该类采用生成-消费的模式，系统在启动的时候，以及监听到目录发生变化
 */
@Service
@Slf4j
public class TravelingFileObserverService {
	@Autowired
	private ACLineSegmentQueryService acLineSegmentQueryService;
	@Autowired
	private SubstationQueryService sqs;
	@Autowired
	private MapDataCache cc;
	@Autowired
	private FaultSystemProperties fp;
	@Autowired
	private SuperDAO dao;
	// 工作线程池
	private ExecutorService worker = Executors.newFixedThreadPool(4);
	// 缓存队列
	private BlockingQueue<FileHandler> fileCache = new LinkedBlockingQueue<>();
	// 合法的文件后缀
	private static final String recordSuffix = "record";
	private static final String equSuffix = "equ";
	private FileAlterationMonitor fileAlterationMonitor = new FileAlterationMonitor();
	// 推送文件路径
	private String pushFilePath = "";
	List<LineSegment> listLineSegment = new ArrayList<LineSegment>();

	@PostConstruct
	public void init() {
		pushFilePath = fp.getTRAVELING_FILE_PATH();
		getLineSegment();
		if(StringUtils.isBlank(pushFilePath)) {
			log.error("行波文件推送文件路径获取失败,请配置行波文件路径;sduept.bigdataserver.fault-module.TRAVELING_FILE_PATH");
			return;
		}
		try {
			initInternal();
		} catch (Exception e) {
			log.error("行波文件推送文件路径获取异常,请配置行波文件路径;sduept.bigdataserver.fault-module.TRAVELING_FILE_PATH");
			return;
		}
		Thread t = new Thread(new Runnable() {
			@Override
			public void run() {
				startInternal();
			}
		}, "TravelingFileObserverService-Thread");
		t.setDaemon(true);
		t.start();
	}

	// 根据参数设置的文件保留期限，删除过期文件
	public void deleteInvalidJSONFile() {
		int day = 10;
		File dir = new File(pushFilePath);
		long current = System.currentTimeMillis();
		for (File f : dir.listFiles()) {
			if (f.getName().contains("comp") && current - f.lastModified() > 86400 * 1000 * day) {
				FileUtils.deleteQuietly(f);
			}
		}
	}
	/**
	 * 每天晚上十点，定期清理10天之前的文件。
	 */
	@Scheduled(cron="0 0 22 * * ?")
	public void deleteInvalidWaveFile() {
		System.out.println("执行了定时删除的方法");
		int day = 10;
		File dir = new File(pushFilePath);
		long current = System.currentTimeMillis();
		for (File f : dir.listFiles()) {
			if (f.getName().contains("comp") && current - f.lastModified() > 86400 * 1000 * day) {
				FileUtils.deleteQuietly(f);
			}
		}
	}

	private void initInternal() {
		FileAlterationObserver fileAlterationObserver = new FileAlterationObserver(pushFilePath);
		FileAlterationListener fileAlterationListener = new FileAlterationReload();
		// 注册监听器
		fileAlterationObserver.addListener(fileAlterationListener);
		// 注册观察者
		fileAlterationMonitor.addObserver(fileAlterationObserver);
		// 初始化已经存在的文件
		File dir = new File(pushFilePath);
		for (File f : dir.listFiles()) {
			if (!f.isFile()) {
				continue;
			}
			String fileName = f.getName();
			String suffix = fileName.substring(fileName.lastIndexOf(".") + 1);
			if (recordSuffix.equals(suffix)) {
				fileCache.add(new RecordFileHandler(f));
			} else if (equSuffix.equals(suffix)) {
				fileCache.add(new EquFileHandler(f));
			}
		}
	}

	/**
	 * 内部启动，主要负责在缓冲队列中获取未处理的文件对象，并进行解析入库
	 */
	private void startInternal() {
		// 启动监听
		try {
			fileAlterationMonitor.start();
		} catch (Exception e) {
			e.printStackTrace();
		}
		while (true) {
			FileHandler f = null;
			try {
				f = fileCache.take();
			} catch (InterruptedException e) {
				log.info(Thread.currentThread() + ":Interrupted");
			}
			if (f != null) {
				worker.execute(f);
			}
		}
	}
    /**
     * 获取缓存中的所有线路数据（交流直流）
     * @return
     */
	private void getLineSegment() {
		Collection<DcLinesegment> dClist = (Collection<DcLinesegment>) cc.getDclineid_map().values();
		for (DcLinesegment dcLinesegment : dClist) {
			listLineSegment.add((LineSegment) dcLinesegment);
		}
		Collection<ACLineSegment> aClist = (Collection<ACLineSegment>) cc.getAclineid_map().values();
		for (ACLineSegment acLinesegment : aClist) {
			listLineSegment.add((LineSegment) acLinesegment);
		}
	}

	/**
	 * 文件监听类，当监听文件夹发生变化时候，触发响应的事件
	 * 
	 * @author wk
	 *
	 */
	private class FileAlterationReload extends FileAlterationListenerAdaptor {
		@Override
		public void onFileChange(File file) {
			logic(file);
		}

		@Override
		public void onFileCreate(File file) {
			logic(file);
		}

		private void logic(File file) {
			if (!file.isFile()) {
				log.info("该文件是目录" + file.getName());
				return;
			}
			String fileName = file.getName();
			String suffix = fileName.substring(fileName.lastIndexOf(".") + 1);
			if (recordSuffix.equals(suffix)) {
				fileCache.add(new RecordFileHandler(file));
			} else if (equSuffix.equals(suffix)) {
				fileCache.add(new EquFileHandler(file));
			}
		}
	}

	/**
	 * 文件处理抽象类
	 * 
	 * @author wk
	 *
	 * @param <T>
	 */
	private abstract class FileHandler implements Runnable {
		private File file;

		public FileHandler(File file) {
			this.file = file;
		}

		@Override
		public void run() {
			try {
				String context = FileUtils.readFileToString(file, "UTF-8");
				doHandler(context);
				postHandler();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		private void postHandler() {
			String fileName = file.getName() + "_comp";
			String path = pushFilePath + fileName;
			try {
				FileUtils.moveFile(file, new File(path));
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * 解析内容
		 * 
		 * @param context
		 * @return
		 */
		public abstract void doHandler(String context);
	}

	/**
	 * @author wk 推送过来的，设备文件处理 context格式如下： [ { "ops":"操作",//create/delete/update
	 *         "context":{ "id":"设备id", "name":"设备名称", "deviceType":"设备类型",
	 *         "ip":"设备ip", "ipSlave":"备用ip", "manufacturer":"生产厂家",
	 *         "monitorLines":"监控线路", "provinceName":"省份名称", "cityName":"城市名称",
	 *         "areaName":"区域名称", "runTime":"投运日期", "stationName":"变电站名称",
	 *         "updateTime":"更新日期" } },{.....} ]
	 */
	private class EquFileHandler extends FileHandler {
		public EquFileHandler(File file) {
			super(file);
		}

		@Override
		public void doHandler(String context) {
			JSONArray jsonArray = JSONArray.parseArray(context);
			for (Iterator<Object> it = jsonArray.iterator(); it.hasNext();) {
				JSONObject jsonObject = (JSONObject) it.next();
				String value = jsonObject.getString("ops");
				if ("delete".equals(value)) {
					String string = jsonObject.getString("context");
					String deleteId = JSONObject.parseObject(string).getString("id");
					dao.delete(deleteId, TravelingWaveEquipment.class);
				} else if ("create".equals(value) || "update".equals(value)) {
					String string = jsonObject.getString("context");
					TravelingWaveEquipmentTemp equ = JSONObject.parseObject(string, TravelingWaveEquipmentTemp.class);
					TravelingWaveEquipment equment = changeModel(equ);
					wrapper(equment);
					dao.createOrUpdate(equment, TravelingWaveEquipment.class);
				}
			}
		}

		private TravelingWaveEquipment changeModel(TravelingWaveEquipmentTemp equ) {
			TravelingWaveEquipment travelingWaveEquipment = new TravelingWaveEquipment();
			travelingWaveEquipment.setId(equ.getId());
			travelingWaveEquipment.setName(equ.getName());
			travelingWaveEquipment.setDeviceType(equ.getDeviceType());
			travelingWaveEquipment.setIp(equ.getIp());
			travelingWaveEquipment.setIpSlave(equ.getIpSlave());
			travelingWaveEquipment.setManufacturer(equ.getManufacturer());
			travelingWaveEquipment.setMonitorLines(equ.getMonitorLines());
			travelingWaveEquipment.setAreaName(equ.getAreaName());
			travelingWaveEquipment.setCityName(equ.getCityName());
			travelingWaveEquipment.setProvinceName(equ.getProvinceName());
			travelingWaveEquipment.setRunTime(equ.getRunTime());
			travelingWaveEquipment.setSubName(equ.getStationName());
			return travelingWaveEquipment;
		}

		/**
		 * 补充完善传入参数，比如：只有stationName，但是无stationId，需要通过该方法补充
		 * 
		 * @param equ
		 */
		private void wrapper(TravelingWaveEquipment equ) {
			List<Substation> lists = cc.getS500List();
			Substation s = sqs.getSubstationByName(equ.getSubName(), cc.getS500List());
			String lineStr = equ.getMonitorLines();
			String[] b = lineStr.split(",");
			String linesId = "";
			for (int i = 0; i < b.length; i++) {
				LineSegment acLine = acLineSegmentQueryService.getLineSegmentByName(b[i], listLineSegment);
				if (acLine != null) {
					if (i < b.length - 1) {
						linesId = linesId + acLine.getId() + ",";
					} else {
						linesId = linesId + acLine.getId();
					}
				}
			}
			equ.setMonitorLines(linesId);
			if (s != null) {
				equ.setSubId(s.getId());
			}
		}

	}

	/**
	 * 推送过来的故障记录文件处理类
	 * 
	 * @author wk
	 *
	 */
	private class RecordFileHandler extends FileHandler {

		public RecordFileHandler(File file) {
			super(file);
		}

		public void doHandler(String context) {
			List<TravelingWaveFaultRecordTemp> list = JSONArray.parseArray(context, TravelingWaveFaultRecordTemp.class);
			
			List<TravelingWaveFaultRecord> listRecord = new ArrayList<TravelingWaveFaultRecord>(); 
			
			wrapper(list,listRecord);
			
			
			
			for (TravelingWaveFaultRecord r : listRecord) {
				dao.createOrUpdate(r, TravelingWaveFaultRecord.class);
			}
		}

		/**
		 * 补充完善传入参数，比如：只有stationName，但是无stationId，需要通过该方法补充
		 * 
		 * @param equ
		 */
		private void wrapper(List<TravelingWaveFaultRecordTemp> records,List<TravelingWaveFaultRecord> listRecord) {
			Pattern p = Pattern.compile("[^0-9]");
			
			for (TravelingWaveFaultRecordTemp temp : records) {
				TravelingWaveFaultRecord tfr = new TravelingWaveFaultRecord();
				Substation s = sqs.getSubstationByName(temp.getLocalStationName(), cc.getS500List());
				if (s != null) {
					tfr.setLocalStationId(s.getId());
				}
				s = sqs.getSubstationByName(temp.getPeerStationName(), cc.getS500List());
				if (s != null) {
					tfr.setPeerStationId(s.getId());
				}
				LineSegment acLine = acLineSegmentQueryService.getLineSegmentByName(temp.getLineName(), listLineSegment);
				if (acLine != null) {
					tfr.setLineId(acLine.getId());
				}
				if (temp.getWaveVelocity() != null) {
					Matcher m = p.matcher(temp.getWaveVelocity());
					tfr.setWaveVelocity(Double.parseDouble(m.replaceAll("")));
				}
				tfr.setId(temp.getId());
				tfr.setName(temp.getName());
				tfr.setPickTime(temp.getPickTime());
				tfr.setSaveTime(temp.getSaveTime());
				tfr.setType(temp.getType());
				tfr.setLineName(temp.getLineName());
				tfr.setLocalIedName(temp.getLocalIedName());
				tfr.setLocalDeviceId(temp.getLocalDeviceId());
				tfr.setLocalDeviceName(temp.getLocalDeviceName());
				tfr.setLocalDeviceDesc(temp.getLocalDeviceDesc());
				tfr.setLocalStationName(temp.getLocalStationName());
				tfr.setPeerIedName(temp.getPeerIedName());
				tfr.setPeerDeviceId(temp.getPeerDeviceId());
				tfr.setPeerDeviceName(temp.getPeerDeviceName());
				tfr.setPeerDeviceDesc(temp.getPeerDeviceDesc());
				tfr.setPeerStationName(temp.getPeerStationName());
				tfr.setLocalFaultResultNumber(temp.getLocalFaultResultNumber());
				tfr.setLocalFaultResultDesc(temp.getLocalFaultResultDesc());
				tfr.setLocalDeviceFaultResultNumber(temp.getLocalDeviceFaultResultNumber());
				tfr.setLocalDeviceFaultResultDesc(temp.getLocalDeviceFaultResultDesc());
				tfr.setPeerFaultResultDesc(temp.getPeerFaultResultDesc());
				tfr.setPeerFaultResultNumber(temp.getPeerFaultResultNumber());
				tfr.setPeerDeviceFaultResultNumber(temp.getPeerDeviceFaultResultNumber());
				tfr.setPeerDeviceFaultResultDesc(temp.getPeerDeviceFaultResultDesc());
				tfr.setFileState(temp.getFileState());
				tfr.setFileStateStr(tfr.getFileStateStr());
				tfr.setLocalWaveTimeNs(temp.getLocalWaveTimeNs());
				tfr.setPeerWaveTimeNs(temp.getPeerWaveTimeNs());
				tfr.setLocalSecWaveTime(temp.getLocalSecWaveTime());
				tfr.setLocalXbFileName(temp.getLocalXbFileName());
				tfr.setPeerXbFileName(temp.getPeerXbFileName());
				tfr.setLocalLbqFileName(temp.getLocalLbqFileName());
				tfr.setPeerLbqFileName(temp.getPeerLbqFileName());
				listRecord.add(tfr);
			}
		}
	}

	/**
	 * "id":"设备id", 
	 * "name":"设备名称", 
	 * "deviceType":"设备类型",  
	 * "ip":"设备ip",
	 * "ipSlave":"备用ip", 
	 * "manufacturer":"生产厂家", 
	 * "monitorLines":"监控线路", 
	 * "provinceName":"省份名称",  
	 * "cityName":"城市名称", 
	 * "areaName":"区域名称", 
	 * "runTime":"投运日期", 
	 * "stationName":"变电站名称", 
	 * "updateTime":"更新日期"
	 * 字段表不同，中间转换模型
	 * @author lixz
	 */
	@Getter
	@Setter
	private static class TravelingWaveEquipmentTemp extends IdentifiedObject {
		/**
		* 
		*/
		private static final long serialVersionUID = 4480425665132387145L;
		/**
		 * 设备ip
		 */
		private String ip;
		/**
		 * 设备备用ip
		 */
		private String ipSlave;
		/**
		 * 设备类型
		 */
		private String deviceType;
		/**
		 * 生产厂家
		 */
		private String manufacturer;
		/**
		 * 监控线路，以英文 “,”分割
		 */
		private String monitorLines;
		/**
		 * 省名称
		 */
		private String provinceName;
		/**
		 * 城市名称
		 */
		private String cityName;
		/**
		 * 区域名称
		 */
		private String areaName;
		/**
		 * 投运时间
		 */
		private Date runTime;

		/**
		 * 厂站名称
		 */
		private String stationName;

	}
	@Getter
	@Setter
	private static class TravelingWaveFaultRecordTemp extends IdentifiedObject {
		
		/**
		 * 
		 */
		private static final long serialVersionUID = -2660280273524330042L;
		/**
		 * 故障时间 ，仅仅是二区推送过来的时间，注：不一定是故障
		 */
		private Date pickTime;
		/**
		 * 保存时间
		 */
		private Date saveTime;
		/**
		 * 0表示双端测距1表示单端测距，如果是单端只参考本端数据，对端都为空
		 */
		private short type;
		/**
		 * 线路名称
		 */
		private String lineName;
		/**
		 * 线路标识
		 */
		private String lineId;
		
		/**
		 * 本端设备ied
		 */
		private String localIedName;
		/**
		 * 本端设备唯一标识
		 */
		private String localDeviceId;
		/**
		 * 本端设备名称
		 */
		private String localDeviceName;
		/**
		 * 本端设备厂家及型号描述
		 */
		private String localDeviceDesc;
		/**
		 * 本端变电站名称
		 */
		private String localStationName;
		/**
		 * 本端变电站id
		 */
		private String localStationId;
		
		/**
		 * 对端设备ied
		 */
		private String peerIedName;
		/**
		 * 对端设备唯一标识
		 */
		private String peerDeviceId;
		/**
		 * 对端设备名称
		 */
		private String peerDeviceName;
		/**
		 * 对端设备厂家及型号描述
		 */
		private String peerDeviceDesc;
		/**
		 * 对端变电站名称
		 */
		private String peerStationName;
		/**
		 * 对端变电站id
		 */
		private String peerStationId;
		/**
		 * 波速度
		 */
		private String  waveVelocity;
		/**
		 * 距本端站测距故障结果（浮点表示）
		 */
		private Double  localFaultResultNumber;
		/**
		 *距本端站测距故障结果（字符串表示）
		 */
		private String localFaultResultDesc;
		/**
		 * 距本端站设备测距故障结果（浮点表示）
		 */
		private Double  localDeviceFaultResultNumber;
		/**
		 * 距本端站设备测距故障结果（字符串表示）
		 */
		private String localDeviceFaultResultDesc;
		
		/**
		 * 距对端站测距故障结果（浮点表示）
		 */
		private Double  peerFaultResultNumber;
		/**
		 *距对端站测距故障结果（字符串表示）
		 */
		private String peerFaultResultDesc;
		/**
		 * 距对端站设备测距故障结果（浮点表示）
		 */
		private Double  peerDeviceFaultResultNumber;
		/**
		 * 距对端站设备测距故障结果（字符串表示）
		 */
		private String peerDeviceFaultResultDesc;
		
		/**
		 * 文件上传状态
		 */
		private short fileState;
		/**
		 * 文件上传状态描述
		 */
		private String fileStateStr;
		/**
		 * 本端首波头纳秒时间戳
		 */
		private long localWaveTimeNs;
		/**
		 * 对端首波头纳秒时间戳
		 */
		private long peerWaveTimeNs;
		/**
		 * 本端二次波头纳秒时间戳
		 */
		private long localSecWaveTime;
		/**
		 * 本端行波文件名字
		 */
		private String localXbFileName;
		/**
		 * 对端行波文件名字
		 */
		private String peerXbFileName;
		/**
		 * 本端工频文件名字
		 */
		private String localLbqFileName;
		/**
		 * 对端工频文件名字
		 */
		private String peerLbqFileName;
		/**
		 * 故障记录的id（cim_fault_record）
		 */
		private String faultId;
	}
}
